package com.audaque.springboot.foshanupload.kafka.filter;

/**
 * @author zgb
 * @desc ...
 * @date 2023-04-14 15:17:29
 */

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Component;

/**
 * 消息过滤器
 */
@Component
public class MyKafkaFilter {

    @Autowired
    ConsumerFactory<Object, Object> consumerFactory;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<Object, Object> listenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        // 设置被过滤的消息将被丢弃
        factory.setAckDiscarded(true);
        // 设置消息过滤的策略
        factory.setRecordFilterStrategy(consumerRecord -> {
            if (consumerRecord.value().toString().length() < 5) {
                // 返回 false 则消息不被过滤
                return false;
            }
            // 返回 true 则消息被过滤
            System.out.println("消息不合格~被过滤丢弃~");
            return true;
        });
        return factory;
    }

}
